package com.sts.sink.impl;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.StrUtil;
import com.sts.jdbc.JdbcOperation;
import com.sts.jdbc.model.ColumnInfo;
import com.sts.sink.model.JdbcSinkConfig;
import com.sts.source.StsSource;
import com.sts.task.StsTask;

import java.math.BigDecimal;
import java.sql.PreparedStatement;
import java.sql.Time;
import java.sql.Timestamp;
import java.sql.Types;
import java.util.List;
import java.util.Map;

public class MysqlSinkImpl extends JdbcSinkImpl {

    public MysqlSinkImpl(StsSource stsSource, JdbcSinkConfig jdbcSinkConfig, StsTask stsTask, JdbcOperation jdbcOperation) {
        super(stsSource, jdbcSinkConfig, stsTask, jdbcOperation);
    }

    @Override
    public void insertSetCellValue(PreparedStatement preparedStatement, List<ColumnInfo> columnInfoList, List<String> rowValueList) throws Exception {
        int size = columnInfoList.size();

        for (int i = 0; i < size; i++) {
            String typeName = columnInfoList.get(i).getTypeName();
            String columnValue = null;

            int rowSize = rowValueList.size();
            if (i < rowSize) {
                columnValue = rowValueList.get(i);
            }

            int index = i + 1;
            if (StrUtil.isNotEmpty(columnValue)) {
                setPreparedStatement(preparedStatement, typeName, index, columnValue);
            } else {
                preparedStatement.setNull(index, Types.NULL);
            }
        }
    }

    @Override
    public String buildUpsertSql(List<ColumnInfo> columnInfoList, Map<String, ColumnInfo> tableAllPrimaryKeyColumnMap) {
        // INSERT INTO table_name (id, column1, column2) VALUES (1, 'value1', 'value2') ON DUPLICATE KEY UPDATE column1='new_value1', column2='new_value2';

        // INSERT INTO `test`.`student` (`id`, `name`, `age`, `sex`, `birthday`, `score`) VALUES (?, ?, ?, ?, ?, ?)
        // on duplicate key update `name` = ?, `age` = ?, `sex` = ?, `birthday` = ?, `score` = ?

        // 该语句会先执行 INSERT 操作，若 INSERT 成功，则后面的 UPDATE 语句会被忽略，若插入时发生主键或唯一索引冲突，则执行 UPDATE 操作
        String identifier = getJdbcSinkConfig().getJdbcConnectionInfo().getIdentifier();

        StringBuilder upsertSql = new StringBuilder("INSERT INTO ");
        upsertSql.append(identifier).append(getJdbcSinkConfig().getSchema()).append(identifier)
                .append(".")
                .append(identifier).append(getJdbcSinkConfig().getTable()).append(identifier)
                .append(" (");

        for (ColumnInfo columnInfo : columnInfoList) {
            upsertSql.append(identifier).append(columnInfo.getColumnName()).append(identifier).append(", ");
        }

        upsertSql.delete(upsertSql.lastIndexOf(","), upsertSql.length());
        upsertSql.append(") VALUES (");

        upsertSql.append("?, ".repeat(columnInfoList.size()));

        upsertSql.delete(upsertSql.lastIndexOf(","), upsertSql.length());
        upsertSql.append(") on duplicate key update ");

        for (ColumnInfo columnInfo : columnInfoList) {
            String columnName = columnInfo.getColumnName();

            // 不是主键的字段需要拼接在 on duplicate key update 后面
            if (!tableAllPrimaryKeyColumnMap.containsKey(columnName.toUpperCase())) {
                upsertSql.append(identifier).append(columnName).append(identifier).append(" = ?, ");
            }
        }

        upsertSql.delete(upsertSql.lastIndexOf(","), upsertSql.length());
        return upsertSql.toString();
    }

    @Override
    public void upsertSetCellValue(PreparedStatement preparedStatement, List<ColumnInfo> columnInfoList, List<String> rowValueList, Map<String, ColumnInfo> tableAllPrimaryKeyColumnMap) throws Exception {
        int size = columnInfoList.size();

        for (int i = 0; i < size; i++) {
            String typeName = columnInfoList.get(i).getTypeName();
            String columnValue = null;

            int rowSize = rowValueList.size();
            if (i < rowSize) {
                columnValue = rowValueList.get(i);
            }

            int index = i + 1;

            if (StrUtil.isNotEmpty(columnValue)) {
                setPreparedStatement(preparedStatement, typeName, index, columnValue);
            } else {
                preparedStatement.setNull(index, Types.NULL);
            }
        }

        int count = 0;
        for (int i = 0; i < size; i++) {
            ColumnInfo columnInfo = columnInfoList.get(i);
            if (tableAllPrimaryKeyColumnMap.containsKey(columnInfo.getColumnName().toUpperCase())) {
                continue;
            }
            String typeName = columnInfo.getTypeName();
            String columnValue = rowValueList.get(i);
            int index = size + count + 1;
            if (StrUtil.isNotEmpty(columnValue)) {
                setPreparedStatement(preparedStatement, typeName, index, columnValue);
            } else {
                preparedStatement.setNull(index, Types.NULL);
            }
            count++;
        }
    }


    public void setPreparedStatement(PreparedStatement preparedStatement, String typeName, int index, String columnValue) throws Exception {
        switch (typeName) {
            case "VARCHAR",
                 "CHAR",
                 "TEXT",
                 "TINYTEXT",
                 "MEDIUMTEXT",
                 "LONGTEXT",
                 "BINARY",
                 "VARBINARY",
                 "BLOB",
                 "TINYBLOB",
                 "MEDIUMBLOB",
                 "LONGBLOB",
                 "YEAR",
                 "JSON",
                 "ENUM",
                 "SET" -> preparedStatement.setString(index, columnValue);
            case "BIT", "TINYINT" -> preparedStatement.setByte(index, Byte.parseByte(columnValue));
            case "INT", "MEDIUMINT" -> preparedStatement.setInt(index, Integer.parseInt(columnValue));
            case "SMALLINT" -> preparedStatement.setShort(index, Short.parseShort(columnValue));
            case "BIGINT" -> preparedStatement.setLong(index, Long.parseLong(columnValue));
            case "FLOAT" -> preparedStatement.setFloat(index, Float.parseFloat(columnValue));
            case "DOUBLE" -> preparedStatement.setDouble(index, Double.parseDouble(columnValue));
            case "DECIMAL" -> preparedStatement.setBigDecimal(index, new BigDecimal(columnValue));
            case "DATE" -> preparedStatement.setDate(index, DateUtil.parse(columnValue).toSqlDate());
            case "BOOLEAN" -> preparedStatement.setBoolean(index, Boolean.parseBoolean(columnValue));
            case "TIME" -> preparedStatement.setTime(index,
                    Time.valueOf(DateUtil.parse(columnValue).toLocalDateTime().toLocalTime())
            );
            case "DATETIME", "TIMESTAMP" -> {
                // yyyy-MM-dd HH:mm:ss.SSS......
                if (columnValue.length() > 26) {
                    columnValue = columnValue.substring(0, 26);
                }
                preparedStatement.setTimestamp(index,
                        Timestamp.valueOf(DateUtil.parse(columnValue).toLocalDateTime())
                );
            }
            default -> throw new RuntimeException("未知数据类型: " + typeName);
        }
    }
}
